package com.sdemo.utils.mq.activemq;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

import com.google.common.collect.Maps;
 

public class ActiveMQConsumer {

	private static final Logger logger = Logger.getLogger(ActiveMQProducer.class);

	private static final String mqUrl = "tcp://127.0.0.1:61616?wireFormat.cacheSize=2048&connectionTimeout=3&soTimeout=1000&tcpNoDelay=true";//ConfigUtil.getString("activemq.url","tcp://172.16.0.104:61616?wireFormat.cacheSize=2048&connectionTimeout=3&soTimeout=1000&tcpNoDelay=true");
	// private static final String url=
	// private static final String user ="admin";//
	// ActiveMQConnectionFactory.DEFAULT_USER;
	private static final String user ="admin";// ConfigUtil.getString("ActiveUser");// ActiveMQConnectionFactory.DEFAULT_USER;
	// private static final String password =
	// "admin";//ActiveMQConnectionFactory.DEFAULT_PASSWORD;
	private static final String password = "admin";//ConfigUtil.getString("ActivePwd");;// ActiveMQConnectionFactory.DEFAULT_PASSWORD;

	/**
	 * 创建Map
	 */
	private static final Map<String, ActiveMQConsumer> instMap = Maps.newConcurrentMap();
	/**
	 * 
	 */
	protected ActiveMQConnectionFactory factory;
	/**
	 * 
	 */
	private Connection connection;
	/**
	 * 
	 */
	private Session session;
	/**
	* 
	*/
	private MessageConsumer consumer;

	private String myTarget;

	/**
	 * 获取实例
	 * 
	 * @param <T>
	 * 
	 * @param target
	 * @param mode
	 * @return
	 */
	public static <T> ActiveMQConsumer getInstance(String target, String clientID, JMSMode mode,MsgHandler<T> handler) {
		if (null != mode && null != target && null != clientID) {
			String key = target + clientID + mode;
			if (!instMap.containsKey(key)) {
				synchronized (instMap) {
					if (!instMap.containsKey(key)) {
						ActiveMQConsumer consumer = new ActiveMQConsumer();
						consumer.init(clientID);
						consumer.createConsumer(mode, target, handler);
						instMap.put(key, consumer);
					}
				}
			}
			return instMap.get(key);
		} else {
			logger.error("参数错误");
			return null;
		}
	}

	/**
	 * 创建消费者
	 * 
	 * @param <T>
	 */
	private <T> void createConsumer(JMSMode jmsMode, String target, final MsgHandler<T> handler) {

		this.myTarget = jmsMode.toString() + "." + target;

		if (null != session) {
			try {
				Destination dest = null;
				switch (jmsMode) {
				case TOPIC:
					dest = session.createTopic(target);
					break;
				case QUEUE:
					dest = session.createQueue(target);
					break;
				}
				consumer = session.createConsumer(dest);
				consumer.setMessageListener(new MessageListener() {

					@Override
					public void onMessage(Message message) {
						if (message instanceof BytesMessage) {
							BytesMessage msg = (BytesMessage) message;
							try {
								byte[] buf = new byte[(int) msg.getBodyLength()];
								msg.readBytes(buf);
								T t = handler.deserializer(buf);
								if (null != t) {
									handler.onMsg(t);
								}
							} catch (JMSException e) {
								logger.error("消息处理异常", e);
							}
						}
						if (message instanceof TextMessage) {
							TextMessage msgs = (TextMessage) message;
							try {
								T t = handler.deserializer(msgs.getText().getBytes("UTF-8"));
								if (null != t) {
									handler.onMsg(t);
								}
							} catch (Exception e) {
								logger.error("TextMessage消息处理异常", e);
							}
						}
					}
				});
				connection.start();

			} catch (JMSException e) {
				logger.error(myTarget + "创建失败", e);
			}
		} else {
			logger.error("Session 为空创建生产者失败");
		}
	}

	/**
	 * 初始化
	 * 
	 * @param clientID
	 */
	private void init(String clientID) {

		if (null == connection) {
			try {
				// failover:(tcp://primary:61616)?timeout=3000
				String brokerURL = "failover:(" + mqUrl + ")";
				logger.info(" ready connect to activemq: " + brokerURL);
				factory = new ActiveMQConnectionFactory(brokerURL);
				factory.setUserName(user);
				factory.setPassword(password);
				// ACK配置
				// factory.setOptimizeAcknowledge(true);
				// factory.setUseAsyncSend(true);
				// factory.setUseCompression(true);
				// 预取配置
				Properties props = new Properties();
				props.setProperty("prefetchPolicy.queuePrefetch", "1");
				props.setProperty("prefetchPolicy.queueBrowserPrefetch", "1");
				props.setProperty("prefetchPolicy.durableTopicPrefetch", "100");
				props.setProperty("prefetchPolicy.topicPrefetch", "32766");
				// factory.setProperties(props);
				// // factory.setAlwaysSyncSend(true);
				// // //
				// // factory.setNonBlockingRedelivery(true);
				// // factory.setSendAcksAsync(true);
				// // // 预读
				// // ActiveMQPrefetchPolicy prefetchPolicy = new
				// ActiveMQPrefetchPolicy();
				// // prefetchPolicy.setQueuePrefetch(15000);
				// // factory.setPrefetchPolicy(prefetchPolicy);
				// 重发策略
				RedeliveryPolicy queuePolicy = new RedeliveryPolicy();
				queuePolicy.setInitialRedeliveryDelay(0);
				queuePolicy.setRedeliveryDelay(1000);// 重发延时
				queuePolicy.setUseExponentialBackOff(false);
				queuePolicy.setMaximumRedeliveries(3);// 重发次数
				factory.setRedeliveryPolicy(queuePolicy);
				// 创建连接
				connection = factory.createConnection();
				if (StringUtils.isNotBlank(clientID)) {
					try {
						clientID = getClientIDString(clientID);
						logger.info("connect activemq user ClientID:" + clientID);
						connection.setClientID(clientID);
					} catch (JMSException e) {
						logger.error("削费者客户端ID：" + clientID + "失败");
					}
				}
			} catch (JMSException e) {
				logger.error("ActiveMq创建连接失败", e);
			}
		}
		if (null != connection) {
			if (null == session) {
				try {
					session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
				} catch (JMSException e) {
					logger.error("ActiveMq创建会话失败", e);
				}
			}
		} else {
			logger.error("连接为空，无法创建Session");
		}

	}

	/**
	 * 获取客户端ID
	 * 
	 * @return
	 */
	private String getClientIDString(String clientID) {
		String ip = "UnknowHost";
		try {
			ip = InetAddress.getLocalHost().getHostAddress();
		} catch (UnknownHostException e) {
			logger.error("获取本机IP失败", e);
		}
		return clientID + "@" + ip + "-" + System.currentTimeMillis();
	}

	/**
	 * 关闭连接
	 */
	public static void shutdown() {
		Iterator<ActiveMQConsumer> ite = instMap.values().iterator();
		ActiveMQConsumer inst = null;
		while (ite.hasNext()) {

			inst = ite.next();
			if (null != inst.consumer) {
				try {
					inst.consumer.close();
				} catch (JMSException e) {
					logger.error(inst.myTarget + "关闭失败", e);
				}
			}
		}
		if (inst == null) {
			return;
		}
		if (null != inst.session) {
			try {
				inst.session.close();
			} catch (JMSException e) {
				logger.error("关闭Session异常", e);
			}
		}
		if (null != inst.connection) {
			try {
				inst.connection.close();
			} catch (JMSException e) {
				logger.error("关闭Connect异常", e);
			}
		}
	}

}
